/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.sss.common.SentinelThreadPool;
import org.sss.common.io.IOThread;
import org.sss.common.io.KeyFormatter;
import org.sss.common.io.StreamDB;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;

/**
 * Server side of 'lstuplegroup.sh' implementation.
 */
public class CommandGetAllTupleGroups {
  private static Logger logger = SssServer.getLogger(CommandGetAllTupleGroups.class);

  public static Set<GroupID> execute(SentinelThreadPool iotp, List<StorageNode> storageNodes)
    throws SssException, InterruptedException {
    final Set<GroupID> allGroups = new HashSet<GroupID>();
    final CountDownLatch latch = new CountDownLatch(storageNodes.size());
    final List<Throwable> exps = new ArrayList<Throwable>();

    for (final StorageNode node: storageNodes) {
      iotp.execute(new Runnable() {
        @Override
        public void run() {
          try {
            final IOThread iot = IOThread.currentThread();
            if (iot.isStreamDB()) {
              final StreamDB db = iot.getStreamDB(node);
              synchronized (allGroups) {
                allGroups.addAll(db.listTupleGroups());
              }
            }
            else {
              Set<GroupID> gids = new HashSet<GroupID>();
              byte[] sentinel = { KeyFormatter.SENTINEL };
              byte[] first = { 0 };

              while (true) {
                //
                // this use case of misc.range command is okay for vanilla TT.
                //
                List<byte[]> kvs = iot.getDB(node).range(first, sentinel, 1, true);
                if (kvs.size() != 2) {
                  break;
                }
                GroupID gid = KeyFormatter.getGroupID(kvs.get(0));
                gids.add(gid);
                first = KeyFormatter.getGroupIDSentinel(gid);
              }
              synchronized (allGroups) {
                allGroups.addAll(gids);
              }
            }
          }
          catch (Throwable e) {
            logger.error("", e);
            synchronized (exps) {
              exps.add(e);
            }
          }
          finally {
            latch.countDown();
          }
        }
      });
    }
    latch.await();

    if (!exps.isEmpty()) {
      throw new SssException(exps.get(0));
    }

    return allGroups;
  }
}
